package com.roy.wordcount;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class StormWordCountTopology {

    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException, InterruptedException {
        //1、创建拓扑结构
        TopologyBuilder builder = new TopologyBuilder();
        //创建spout 会接收nimbus发送的任务，
        builder.setSpout("wordCountSpout",new WordCountSpout(),1);
        //创建splitBolt,对接spout，进行单词切割。按照随机分组
        builder.setBolt("mySplitBolt",new SplitBolt(),10).shuffleGrouping("wordCountSpout");
        //创建countBolt，对接splitBolt，进行单词计数，按照word字段分组。这个word要是splitBolt的输出字段
        builder.setBolt("myCountBolt",new CountBolt(),2).fieldsGrouping("mySplitBolt",new Fields("word"));

        //2、创建一个Config
        Config config = new Config();
        config.setNumWorkers(2);

        //3、提交任务： 集群提交和本地提交
        //集群模式：集群模式不需要关闭程序
        //StormSubmitter.submitTopology("myWordCount",config,builder.createTopology());
        //本地模式：本地模式可以在运行完成后关闭程序。
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("word-count", config, builder.createTopology());

        Thread.sleep(100000);

        cluster.shutdown();
    }
}
